package main

import (
	"../dmap"
	"context"
	"github.com/valyala/fasthttp"
	"io/ioutil"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

type Router struct {
	mu                  sync.RWMutex
	client              *http.Client
	baseBackends        []string
	cacheBackends       map[string]string
	mapping             *dmap.EtcdDistributeMap
	timeout             int
	GetKeyFunc          func(*fasthttp.RequestCtx) string
	queryAllOnCacheMiss bool
	queryBackends       bool
}

func KeyAsParamFileName(ctx *fasthttp.RequestCtx) (ret string) {
	if ctx.Request.Header.IsGet() {
		b := ctx.QueryArgs().Peek("filename")
		if len(b) != 0 {
			ret = string(b)
		}
	}
	return
}

func KeyAsDefault(ctx *fasthttp.RequestCtx) string {
	return string(ctx.Path())
}
func KeyAsLast(ctx *fasthttp.RequestCtx) string {
	s := strings.Split(string(ctx.Path()), "/")
	if len(s) > 1 {
		return s[len(s)-1]
	}
	return ""
}
func KeyAsSecond(ctx *fasthttp.RequestCtx) string {
	if s := strings.Split(string(ctx.Path()), "/"); len(s) > 2 {
		return s[2]
	}
	return ""
}

func (r *Router) init() error {
	r.client = &http.Client{
		Timeout: time.Duration(r.timeout) * time.Second,
	}
	if r.mapping != nil {
		if err := r.mapping.Init(); err != nil {
			return err
		}
		if err := r.mapping.Reload(); err != nil {
			return err
		}
		go r.mapping.Watch()
	}
	return nil
}

//as default return all cached backeds
func (r *Router) addrFromCache(incomeCtx *fasthttp.RequestCtx) (ret []string, hit bool) {
	if r.GetKeyFunc == nil || len(r.cacheBackends) == 0 {
		return
	}
	key := r.GetKeyFunc(incomeCtx)
	LOG.Infof("[%d] key is %s", incomeCtx.ConnID(), key)
	var addr string
	if r.mapping != nil {
		id, _, _ := r.mapping.Query(key)
		addr, hit = r.cacheBackends[id]
	}
	if key == "" || r.mapping == nil || !hit {
		hit = false
		for _, v := range r.cacheBackends {
			ret = append(ret, v)
		}
		return
	}
	ret = append(ret, addr)
	return
}

func (r *Router) handler(incomeCtx *fasthttp.RequestCtx) {
	t := time.Now().Nanosecond() / 1000 / 1000
	defer func() {
		duration := time.Now().Nanosecond()/1000/1000 - t
		LOG.Infof("[%d] handle end %s ,costs: %d (ms)\n", incomeCtx.ConnID(), incomeCtx.Request.URI().String(), duration)
	}()
	addrs, hit := r.addrFromCache(incomeCtx)
	if !hit && !r.queryAllOnCacheMiss {
		addrs = []string{}
	}
	if r.queryBackends {
		addrs = append(addrs, r.baseBackends...)
	}
	LOG.Infof("[%d] handle start uri: %s, request type: %s ", incomeCtx.ConnID(), string(incomeCtx.RequestURI()), string(incomeCtx.Request.Header.Method()))
	LOG.Infof("[%d] cache hit: %v, query for addrs %v ", incomeCtx.ConnID(), hit, addrs)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*time.Duration(r.timeout))
	ch := r.DoAndWaitAll(incomeCtx, addrs, cancel)
	var result *RespWithAddr
	actionType := "DEFAULT"
	select {
	case result = <-ch:
		if result.forced {
			actionType = "FORCED"
		} else {
			actionType = "NORMAL"
		}
		LOG.Noticef("[%d] %s resp from %s %d\n", incomeCtx.ConnID(), actionType, result.addr, result.resp.StatusCode)
		if result.resp.ContentLength == 0 {
			LOG.Errorf("[%d] %s head content-length is 0\n", incomeCtx.ConnID(), result.addr)
		}
		cancel()
	case <-ctx.Done():
		LOG.Noticef("[%d] DEFAULT resp 404\n", incomeCtx.ConnID())
	}
	if result != nil {
		body, err := ioutil.ReadAll(result.resp.Body)
		defer result.resp.Body.Close()
		if err == nil {
			incomeCtx.Response.SetStatusCode(result.resp.StatusCode)
			incomeCtx.Response.SetBody(body)
			l := result.resp.Header.Get("Content-Length")
			ll, _ := strconv.Atoi(l)
			incomeCtx.Response.Header.SetContentLength(ll)
			for k, v := range result.resp.Header {
				incomeCtx.Response.Header.Set(k, strings.Join(v, ","))
			}
			return
		}
	}
	incomeCtx.Response.SetStatusCode(http.StatusNotFound)
}

func (r *Router) DoAndWaitAll(originCtx *fasthttp.RequestCtx, dests []string, cancel context.CancelFunc) <-chan *RespWithAddr {
	ch := make(chan *RespWithAddr)
	if len(dests) == 0 {
		cancel()
		return ch
	}
	counter := int32(len(dests))
	for _, dest := range dests {
		req := convertReq(originCtx, dest)
		if req == nil {
			continue
		}
		go r.DoOne(originCtx.ConnID(), req, &counter, ch)
	}
	return ch
}

type RespWithAddr struct {
	resp   *http.Response
	addr   string
	forced bool
}

func convertReq(ctx *fasthttp.RequestCtx, host string) *http.Request {
	m := string(ctx.Method())
	uri := ctx.URI()
	uri.SetHost(host)
	req, err := http.NewRequest(m, uri.String(), nil)
    req.Header.Set("Content-Type",string(ctx.Request.Header.ContentType()))
	if err == nil {
		return req
	}
	return nil
}

func (r *Router) DoOne(id uint64, req *http.Request, counter *int32, output chan *RespWithAddr) {
	resp, err := r.client.Do(req)
	if err != nil {
		LOG.Errorf("[%d] err while get %s , err %v", id, req.Host, err)
		return
	}
	code := resp.StatusCode
	if code == http.StatusOK || code == http.StatusPartialContent { //206 {
		result := &RespWithAddr{
			addr: req.Host,
			resp: resp,
		}
		output <- result
		return
	} else {
		LOG.Noticef("[%d] req from %s return code is %d\n", id, req.Host, code)
	}
	if atomic.AddInt32(counter, -1) == 0 {
		result := &RespWithAddr{
			addr:   req.Host,
			resp:   resp,
			forced: true,
		}
		output <- result
		return
	}
	return
}
